/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package com.oszero.deliver.server.client.emqx;

import cn.hutool.http.HttpRequest;
import cn.hutool.http.HttpResponse;
import cn.hutool.json.JSONUtil;
import com.oszero.deliver.server.client.emqx.dtos.BatchBulkParam;
import com.oszero.deliver.server.client.emqx.dtos.InsertMqttMsgEsParam;
import com.oszero.deliver.server.enums.EmqxMessageType;
import com.oszero.deliver.server.exception.MessageException;
import com.oszero.deliver.server.model.app.EmqxApp;
import com.oszero.deliver.server.model.dto.common.SendTaskDto;
import com.oszero.deliver.server.model.entity.EmqxMessageRecord;
import com.oszero.deliver.server.util.SnowflakeUtil;
import lombok.Data;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import okhttp3.Credentials;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

@Component
@RequiredArgsConstructor
@Slf4j
public class EmqxClient
{
    /**
     * 批量发送
     * https://docs.emqx.com/zh/emqx/v5.1/admin/api-docs.html#tag/Publish/paths/~1publish~1bulk/post
     */
    private final static String url_publishBatchMessages = "/api/v5/publish/bulk";
    private static final ExecutorService executorService = Executors.newFixedThreadPool(5); // 根据需要调整线程池大小

    public void sendMessage(EmqxApp emqxApp, SendTaskDto sendTaskDto) throws IOException
    {
        List<BatchBulkParam> batchBulkParams = new ArrayList<>();
        List<String> users = sendTaskDto.getUsers();

        Map<String, Object> paramMap = sendTaskDto.getParamMap();
        String message = paramMap.get("message").toString();
        String title = paramMap.containsKey("title")?paramMap.get("title").toString():"";
        String picUrls = paramMap.containsKey("picUrls")?paramMap.get("picUrls").toString():"";
        String creator = paramMap.containsKey("creator")?paramMap.get("creator").toString():"";

        //QoS 0 最多一次
        //QoS 1 至少一次
        //QoS 2 仅一次
        int qos = paramMap.containsKey("qos") ?Integer.parseInt(paramMap.get("qos").toString()) : 0;

        //推送ES的数据集
        List<EmqxMessageRecord> emqxMessageRecords = new ArrayList<>();

        EmqxMessageType messageType = EmqxMessageType.SystemNotice;
        if(!creator.isEmpty())
            messageType = EmqxMessageType.UserChat;

        List<String> esToUserIds = new ArrayList<>();
        for (String user : users)
        {
            EmqxMessageRecord emqxMessageRecordItem = EmqxMessageRecord.builder()
                    .id(Long.toString(SnowflakeUtil.getId()))
                    .title(title)
                    .message(message)
                    .picUrls(picUrls)
                    .messageType(messageType)
                    .toUser(user)
                    .createTime(new Date())
                    .creatorId(creator)
                    .build();
            emqxMessageRecords.add(emqxMessageRecordItem);

            batchBulkParams.add(BatchBulkParam.builder().topic(user).payload(JSONUtil.toJsonStr(emqxMessageRecordItem)).qos(qos).build());

            esToUserIds.add(user);
        }

        if(batchBulkParams.isEmpty())
            throw new MessageException("消息列表为空，无法发送消息");

        //如果有ES地址直接推送
        String messageTypeName = messageType.getName();
        if (!emqxApp.getEsCallbackUrl().isEmpty()) {
            executorService.submit(() -> pushToES(emqxApp, title, creator, message, picUrls, messageTypeName, esToUserIds));
        }

        try (HttpResponse response = HttpRequest.post(emqxApp.getUrl()+url_publishBatchMessages)
                .header("Content-Type", "application/json; charset=utf-8")
                .header("Authorization", Credentials.basic(emqxApp.getAppKey(), emqxApp.getAppSecret()))
                .body(JSONUtil.toJsonStr(batchBulkParams))
                .execute())
        {
            @Data
            class EmqxRespBody {
                private Integer reason_code;
                private String message;
            }

            if(response.getStatus() != 200)
            {
                List<EmqxRespBody> emqxRespBody = JSONUtil.toList(response.body(),EmqxRespBody.class);
                EmqxRespBody first = emqxRespBody.stream().findFirst().get();
                throw new MessageException(sendTaskDto, "返回:"+response.getStatus()+",Emqx批量发送失败,错误"+first.reason_code+":"+first.message);
            }


        } catch (Exception e) {
            throw new MessageException(sendTaskDto, "EMQX平台批量发送MQTT消息失败" + e.getMessage());
        }

    }

    /**
     * 推送到ES
     * @param emqxApp 配置文件
     * @param title 标题
     * @param creator 创建人
     * @param message 消息
     * @param picUrls 图片
     * @param messageTypeName 消息类型
     * @param esToUserIds 目标用户
     */
    private void pushToES(EmqxApp emqxApp, String title, String creator, String message, String picUrls, String messageTypeName, List<String> esToUserIds) {
        InsertMqttMsgEsParam esInsertParam = InsertMqttMsgEsParam.builder()
                .title(title)
                .creatorId(creator)
                .message(message)
                .picUrls(picUrls)
                .messageType(messageTypeName)
                .toUserIds(esToUserIds)
                .build();

        try (HttpResponse response = HttpRequest.post(emqxApp.getEsCallbackUrl() + "/noAuth/insertMqttMsgEs")
                .header("Content-Type", "application/json; charset=utf-8")
                .body(JSONUtil.toJsonStr(esInsertParam))
                .execute()) {

            if (response.getStatus() != 200) {
                log.info("消息推送ES失败：" + response.getStatus() + response.body());
            }

        } catch (Exception ex) {
            log.info("消息推送ES失败：" + ex.getMessage());
        }
    }


}
